package com.cnpc.bds.sdk.sink.kafka.executor;

import com.cnpc.bds.sdk.sink.kafka.handler.AbstractHandler;
import com.cnpc.bds.sdk.sink.kafka.meta.ProducerCfg;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 任务流终点-Kafka-Producer.
 */
@Slf4j
public class DownStreamProducer extends AbstractHandler<byte[], Boolean> {

    private final ProducerCfg producerConfig;

    private final String topic;

    private Producer<byte[], byte[]> kafkaProducer;

    public DownStreamProducer(ProducerCfg config) {
        this.producerConfig = config;
        this.topic = config.getTopicMeta().getTopic();
        init0();
    }

    private void init0() {
        kafkaProducer = new KafkaProducer<>(producerConfig.properties());
    }

    @Override
    public Boolean handle(byte[] msg) {
        send(msg);
        return true;
    }

    private void send(final byte[] msg) {
        kafkaProducer.send(new ProducerRecord<>(topic, msg));
        log.debug("---> sink success!");
    }

}
